package com.foreveross.taskservice.infra.db;

import java.io.IOException;
import java.util.Map;

import org.apache.commons.lang3.StringUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.HConnection;
import org.apache.hadoop.hbase.client.HConnectionManager;
import org.apache.hadoop.hbase.client.HTableInterface;

import com.foreveross.taskservice.infra.cfg.ConfigContainer;

/**
 * hbase连接
 * @author xiangsf   
 * @since 2013-2-16
 */
public class HbaseDriverManager {
	
	protected Log logger= LogFactory.getLog(getClass());
	
	private Configuration conf;

	private HBaseAdmin hbaseAdmin;
	
	private HConnection hTablePool; 
	
	
	private final static int DEFAULT_POOL_MAX_SIZE = 10;
	
	
	private static HbaseDriverManager _instance = null;
	
	private HbaseDriverManager() {
		if(conf==null){
			conf= new Configuration();
			Map<String, String> props = ConfigContainer.getInstance().getProp(HBasePropertiesLoader.class);
			for(String key : props.keySet()){
				conf.set(key, props.get(key));
			}

			try {
				hTablePool=HConnectionManager.createConnection(conf);//注意这个值设置的是每个htable表在pool中的最大值
				hbaseAdmin = new HBaseAdmin(conf);
			} catch (Exception e) {
				e.printStackTrace();
			} 
		}
	}
	
	public synchronized static HbaseDriverManager getInstance(){
		if(_instance == null){
			_instance = new HbaseDriverManager();
		}
		return _instance;
	}
	
	
	public Configuration getConf() {
		return conf;
	}

	/**
	 * 获取HBaseAdmin
	 * @return
	 */
	public HBaseAdmin getHBaseAdmin(){
		 return hbaseAdmin;
	}
	
    /**        
     * 返回htablepool连接池中的一个htable
     * @param tableName
     * @return
     * @throws IOException 
     */
    public synchronized HTableInterface getHtable(String tableName) throws IOException{
    	return (HTableInterface)hTablePool.getTable(tableName);//如果hTablePool对象已经存在，直接取出一个htable
    }
    /**
     * 释放表
     * @param tableName
     * @throws IOException
     */
	public synchronized void releaseHTable(String tableName) throws IOException{
    	if(StringUtils.isNotBlank(tableName)){
    		HTableInterface t = getHtable(tableName);
    		
    		if(t != null) {
    			t.close();
    			//采用上面的关闭方法
    			//hTablePool.putTable(t);
    		}
    	}
    }
	/**
     * 释放表连接
     * @param tableName
     * @throws IOException
     */
	public synchronized void releaseHTable(HTableInterface t){
		if(t != null) {
			try {
				t.close();
			} catch (IOException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
			//采用上面的关闭方法
			//hTablePool.putTable(t);
		}
	}
	 /**
     * 创建一张表
     */
    public synchronized void creatTable(String tableName, String[] familys) throws Exception {
    	if (hbaseAdmin.tableExists(tableName)){
    		if(logger.isDebugEnabled()){
    			logger.debug("hbase表"+tableName+"已经存在!");
        	}
    	}else {
            HTableDescriptor tableDesc = new HTableDescriptor(tableName);
            for(int i=0; i<familys.length; i++){
                tableDesc.addFamily(new HColumnDescriptor(familys[i]));
            }
            hbaseAdmin.createTable(tableDesc);
            if(logger.isDebugEnabled()){
    			logger.debug("已成功创建表："+tableName);
        	}
        } 
    }
    
    /**
     * 删除表
     */
    public synchronized void deleteTable(String tableName) throws Exception {
           hbaseAdmin.disableTable(tableName);
           hbaseAdmin.deleteTable(tableName);
         if(logger.isDebugEnabled()){
   			logger.debug("已成功删除表："+tableName);
       	}
    }
    
    public static void main(String[] arg) throws Exception{
    	//logger.info("抓取引警系统配置文件统一注册......");
    	ConfigContainer.getInstance().register(HBasePropertiesLoader.class);

    	HbaseDriverManager manager = HbaseDriverManager.getInstance();
    	System.out.println(manager.getHtable("CRAWL_PLANE_FREIGHT"));
//    	manager.creatTable("CRAWL_PF_SOURCE_DATA", new String[]{"CONTENTS","EXT_CONTENTS"});
//    	manager.creatTable("CRAWL_PLANE_FREIGHT", new String[]{"planeinfo","agentInfos","cabinInfos","priceInfos"});
//    	HBaseRowModel model = new HBaseRowModel();
//    	model.setRowkey("11222");
//    	model.addCell(new HBaseCellModel("CONTENTS","TTTT",0L,"ddddd"));
//    	new HbaseManager().addRecord("CRAWL_PLANE_FREIGHT", model);
    	//manager.deleteTable("CRAWL_PF_SOURCE_DATA");
    	//manager.deleteTable("CRAWL_PLANE_FREIGHT");
    }
}
